package com.zhuimeng.hbase;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.*;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;

import java.io.IOException;

public class HbaseTest {

    Configuration conf = null;
    Connection conn = null;
    HBaseAdmin admin = null;
    @Before
    public void init() throws IOException{
        conf = HBaseConfiguration.create();
        conf.set("hbase.zookeeper.quorum","hadoop1,hadoop2,hadoop3");
        conf.set("hbase.zookeeper.property.clientPort","2181");
        conn = ConnectionFactory.createConnection(conf);
    }

    @After
    public void destory() throws IOException{
        if (conn != null){
            conn.close();
        }
        if (admin != null){
            admin.close();
        }
    }
    @Test
    public void createTable() throws IOException{
        admin = (HBaseAdmin) conn.getAdmin();
        HTableDescriptor friend = new HTableDescriptor(TableName.valueOf("friend"));
        friend.addFamily(new HColumnDescriptor("friends"));
        admin.createTable(friend);
        System.out.println("表创建成功");
    }

    @Test
    public void addFriends() throws IOException{
        String friend1 = "friend1";
        String friend2 = "friend2";
        String friend3 = "friend3";
        put(friend1,friend2);
        put(friend1,friend3);
        put(friend2,friend1);
        put(friend2,friend3);
        put(friend3,friend1);
        put(friend3,friend2);

    }

    public void put(String rowkey, String friendsId) throws IOException{
        Table friend = conn.getTable(TableName.valueOf("friend"));
        Put put = new Put(Bytes.toBytes(rowkey));
        put.addColumn(Bytes.toBytes("friends"),Bytes.toBytes(friendsId),Bytes.toBytes(friendsId));
        friend.put(put);
        friend.close();
        System.out.println("添加好友成功");
    }

    /**
     * 全表扫描
     */
    @Test
    public void scanAllData() throws IOException {
        HTable friend = (HTable) conn.getTable(TableName.valueOf("friend"));
        Scan scan = new Scan();
        ResultScanner resultScanner = friend.getScanner(scan);
        for (Result result : resultScanner) {
            Cell[] cells = result.rawCells();//获取改⾏的所有cell对象
            for (Cell cell : cells) {
                //通过cell获取rowkey,cf,column,value
                String cf = Bytes.toString(CellUtil.cloneFamily(cell));
                String column = Bytes.toString(CellUtil.cloneQualifier(cell));
                String value = Bytes.toString(CellUtil.cloneValue(cell));
                String rowkey = Bytes.toString(CellUtil.cloneRow(cell));
                System.out.println(rowkey + "----" + cf + "--" + column + "---" + value);
            }
        }
        friend.close();
    }

    @Test
    public void testDel() throws IOException{
        delFriend("friend1","friend3");
    }

    public void delFriend(String rowkey, String friendsId) throws IOException {
        //获取表对象
        HTable friend = (HTable) conn.getTable(TableName.valueOf("friend"));
        //创建查询的get对象
        Get get = new Get(Bytes.toBytes(friendsId));
        get.addFamily(Bytes.toBytes("friends"));
        Delete delete = new Delete(Bytes.toBytes(rowkey));
        delete.addColumn(Bytes.toBytes("friends"),Bytes.toBytes(friendsId));
        friend.delete(delete);
        //执⾏查询
        Result res = friend.get(get);
        Cell[] cells = res.rawCells();//获取改⾏的所有cell对象
        for (Cell cell :
                cells) {
            String key = Bytes.toString(CellUtil.cloneRow(cell));
            if (key.equals(friendsId)){
                delete = new Delete(Bytes.toBytes(friendsId));
                delete.addColumn(Bytes.toBytes("friends"),Bytes.toBytes(rowkey));
                friend.delete(delete);


            }
        }
        friend.close();//关闭表对象资源
    }
}
